/**
 * UDPTrafficSource.java 18.09.2006
 */
package org.dicr.traffic.source;

import java.io.*;
import java.net.*;
import java.util.*;

import javax.management.*;

import org.dicr.util.data.exc.*;
import org.dicr.util.net.*;
import org.slf4j.*;

/**
 * Listen UDP for for traffic data. Data of UDP packet is a text, each line of
 * which must be in format:
 * 
 * <PRE>
 * 
 * address|bytesIn|bytesOu\n
 * 
 * </PRE>
 * 
 * , where
 * <UL>
 * <LI>address - IP[/Mask] address of host</LI>
 * <LI>bytesIn - number of bytes input by host</LI>
 * <LI>bytesOu - number of bytes output by host</LI>
 * </UL>
 * No replies send to packets.
 * 
 * @author <A href='http://dicr.org'>Igor A Tarasov</A>
 * @version 060918
 */
public class UDPTrafficSource extends AbstractTrafficSource implements MBeanRegistration, UDPTrafficSourceMBean {

	private static final Logger log = LoggerFactory.getLogger(UDPTrafficSource.class);

	/** Default JMX ObjectName */
	public static final String DEFAULT_OBJECT_NAME = "org.dicr:service=traffic;type=source;name=udp";

	/** Port to listen */
	protected int port = 8731;

	/** Listener task */
	private UDPListenerTask task = null;

	/**
	 * Constructor
	 */
	public UDPTrafficSource() {
		super();
	}

	/**
	 * Constructor
	 * 
	 * @param aport
	 *            UDP port number
	 */
	public UDPTrafficSource(final int aport) {
		super();
		this.setPort(aport);
	}

	/**
	 * Set port.
	 * 
	 * @param aport
	 *            UDP port number to listen.
	 */
	@Override
	public void setPort(final int aport) {
		if (aport < 0 || aport > 65535) throw new IllegalArgumentException("port: " + aport);
		this.port = aport;
	}

	/**
	 * Return port.
	 * 
	 * @return UDP port number.
	 */
	@Override
	public int getPort() {
		return this.port;
	}

	/**
	 * @see org.dicr.traffic.source.UDPTrafficSourceMBean#isRunning()
	 */
	@Override
	public final boolean isRunning() {
		synchronized (this) {
			return this.task != null;
		}
	}

	/**
	 * @see org.dicr.traffic.source.UDPTrafficSourceMBean#start()
	 */
	@Override
	public final void start() {
		synchronized (this) {
			if (!this.isRunning()) {
				UDPTrafficSource.log.info("starting UDP traffic source");
				this.task = new UDPListenerTask();
				this.task.start();
			} else {
				UDPTrafficSource.log.debug("UDP traffic source already running");
			}
		}
	}

	/**
	 * @see org.dicr.traffic.source.UDPTrafficSourceMBean#stop()
	 */
	@Override
	public final void stop() {
		synchronized (this) {
			if (this.isRunning()) {
				UDPTrafficSource.log.info("stopping UDP traffic source");
				this.task.interrupt();
				this.task = null;
			} else {
				UDPTrafficSource.log.debug("traffic source is not running already");
			}
		}
	}

	/**
	 * @see javax.management.MBeanRegistration#preRegister(javax.management.MBeanServer,
	 *      javax.management.ObjectName)
	 */
	@Override
	public ObjectName preRegister(final MBeanServer server, final ObjectName name) throws Exception {
		return name != null ? name : ObjectName.getInstance(UDPTrafficSource.DEFAULT_OBJECT_NAME);
	}

	/**
	 * @see javax.management.MBeanRegistration#postRegister(java.lang.Boolean)
	 */
	@Override
	public void postRegister(final Boolean registrationDone) {
		if (registrationDone != null && registrationDone.booleanValue() && !this.isRunning()) this.start();
	}

	/**
	 * @see javax.management.MBeanRegistration#preDeregister()
	 */
	@Override
	public void preDeregister() throws Exception {
		if (this.isRunning()) this.stop();
	}

	/**
	 * @see javax.management.MBeanRegistration#postDeregister()
	 */
	@Override
	public void postDeregister() {
		// NOP
	}

	/**
	 * Fire traffic to listeners This method for access to parent fire traffic
	 * from UDPListenerTask.
	 * 
	 * @param traffic
	 */
	protected void processTraffic(final Collection<TrafficElement> traffic) {
		if (traffic == null) throw new IllegalArgumentException("null traffic");
		this.fireTraffic(traffic);
	}

	/**
	 * UDP Listener task
	 */
	protected class UDPListenerTask extends Thread {

		/**
		 * Constructor
		 */
		public UDPListenerTask() {
			super("UDP traffic source");
			this.setDaemon(false);
		}

		/**
		 * @see java.lang.Thread#run()
		 */
		@SuppressWarnings("synthetic-access")
		@Override
		public void run() {
			DatagramSocket sock = null;
			try {
				sock = new DatagramSocket(new InetSocketAddress("0.0.0.0", UDPTrafficSource.this.port));
				while (!this.isInterrupted()) {
					try {
						final byte[] data = new byte[100000];
						final DatagramPacket p = new DatagramPacket(data, data.length);
						sock.receive(p);
						if (p.getLength() < 5) continue;

						final List<TrafficElement> traffic = new ArrayList<>();
						final BufferedReader r = new BufferedReader(new InputStreamReader(new ByteArrayInputStream(
								p.getData(), 0, p.getLength())));
						String line = null;
						for (;;) {
							line = r.readLine();
							if (line == null) break;
							line = line.trim();
							final String[] parts = line.split("\\|");
							if (parts.length != 3) throw new IncorrectDataException("incorrect string received: "
									+ line);
							final Net addr = new Net(parts[0]);
							final long bytesIn = Long.parseLong(parts[1]);
							final long bytesOu = Long.parseLong(parts[2]);

							UDPTrafficSource.log.trace("received traffic: addr=" + addr + ", bytesIn=" + bytesIn
									+ ", bytesOu=" + bytesOu);

							if (bytesIn > 0) {
								final TrafficElement el = new TrafficElement();
								el.setDst(addr.getIp());
								el.setDstMask(addr.getMask());
								el.setBytes(bytesIn);
								traffic.add(el);
							}
							if (bytesOu > 0) {
								final TrafficElement el = new TrafficElement();
								el.setSrc(addr.getIp());
								el.setSrcMask(addr.getMask());
								el.setBytes(bytesOu);
								traffic.add(el);
							}
						}
						r.close();
						if (!traffic.isEmpty()) UDPTrafficSource.this.processTraffic(traffic);
					} catch (final Throwable ex) {
						UDPTrafficSource.log.warn("error handling packet", ex);
					}
				}
			} catch (final Exception ex) {
				UDPTrafficSource.log.warn("error", ex);
			} finally {
				if (sock != null) try {
					sock.close();
				} catch (final Exception ex) {
					// NOP
				}
			}
		}
	}
}
